package icasue.net.socket.thread;

import com.alibaba.fastjson.JSON;
import icasue.net.base.annotations.NotNull;
import icasue.net.base.annotations.Nullable;
import icasue.net.protocol.carfun.common.CommandResponse;
import icasue.net.socket.exception.CommandExecException;
import icasue.net.socket.exception.PredicateException;
import icasue.net.socket.thread.state.ChannelState;
import icasue.net.socket.driver.event.EventCommandDriver;
import icasue.net.socket.driver.providers.CommandDriverManager;
import icasue.net.socket.exception.ClientConnectException;
import icasue.net.socket.exception.ProtocolUnSupportException;
import icasue.net.socket.share.MindManger;
import icasue.net.socket.model.event.PoolMesItem;
import icasue.net.socket.model.register.RegisterInfo;
import icasue.net.protocol.carfun.CarFunProtocol;
import icasue.net.protocol.carfun.commands.Command;
import icasue.net.protocol.carfun.commands.EventCommand;
import icasue.net.socket.thread.state.SocketState;
import icasue.net.socket.utils.Cast;
import icasue.net.socket.utils.LogUtil;
import icasue.net.socket.utils.Refection;
import java.io.IOException;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.lang.reflect.Method;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;

/**
 * @Author: Qiao Hang
 * @CreateDate: 2021/3/18 下午5:09
 * @UpdateDate:
 * @Description: Channel 发起组队的客户端会创建Channel实例，为线程资源，该Channel线程被创建者/加入者分享，可以被多个Socket客户端注册.
 * Channel也管理来自不同
 */
public class Channel implements Runnable {

    //channelId.
    private final Long channelId;
    //channel state, keep working state after create and
    private ChannelState state;
    //当前Channel上加入的客户端连接, 与注册信息identity, registerInfoIdentity -> socketProxy.
    private Map<Object,SocketProxy> proxies;
    //当前Channel上所连接的客户端注册的注册信息, identity -> registerInfo
    private Map<Object, RegisterInfo> registerInfos;
    //消息池 {identity,command}
    private volatile BlockingQueue<PoolMesItem<JSON>> mesPool = new LinkedBlockingQueue();
    //响应协议构造句柄.
    private static final MethodHandle $protocolBuilder;
    //默认响应字符集
    private static final Charset UTF_8 = Charset.forName("UTF-8");

    static {
        Method unReflect = Refection.findDeclareMethod(MethodHandles.Lookup.class, "unreflect", new Class[]{Method.class});
        //protocolBuilder's handle.
        $protocolBuilder = Cast.cast(Refection.methodInvoke(unReflect, Refection.methodLookup,new Object[]{
                Refection.findDeclareMethod(CarFunProtocol.class, "builder", new Class[]{Command.class,Integer.class})}),
                MethodHandle.class);
    }

    private void channelInit(){
        this.proxies = new HashMap<>(1<<3,1.0f);
        this.registerInfos = new HashMap<>(1<<2,1.0f);
        this.mesPool = new LinkedBlockingQueue(Integer.MAX_VALUE);
    }

    public Channel(@NotNull Long channelId){
        this.channelInit();
        this.channelId = channelId;
        //状态为working态.
        this.state = ChannelState.WORKING;
    }

    @Override
    public void run() {
        //channel's state is working state, do message sending.
        while (this.state == ChannelState.WORKING){
            try {
                PoolMesItem<JSON> mesItem = mesPool.poll();

                /*处理MesItem, 构建服务端Protocol, 通知处于BIND态的客户端.*/
                if(mesItem != null){
                    //获取BizCommandDriver
                    EventCommandDriver eventDriver = null;
                    try {
                        eventDriver = CommandDriverManager.chooseEventDriver(mesItem.getCommand());
                    }catch (ProtocolUnSupportException unSupport){
                        /*Command is not exist as a EventCommand,
                        developer should check if designed MDCommand push to EventCommand,
                        actually, no such case, for PoolMesItem required a Command as EventCommand.*/
                        continue;
                    }

                    //处理事件请求，构建事件通知的响应体.
                    CommandResponse<JSON> response = null;
                    try {
                        response = eventDriver.dealEventCommand(
                                mesItem.getCommand(),
                                mesItem.getSocketProxy(),
                                mesItem.getRegisterInfo(),
                                this.channelId,
                                mesItem.getData());
                    }catch (ProtocolUnSupportException | CommandExecException exec){
                        /*EventCommand UnSupport or CommandExecException occur will ignore this message..*/
                        continue;
                    }catch (ClientConnectException connectEx){
                        /*Client connect break, will destroy this client. */
                        mesItem.getSocketProxy().destroy();
                        continue;
                    }catch (PredicateException e){
                        /*Occur PredicateException means predicate error,
                        perhaps channel or socket already been released or socket's state
                        is inValid etc, all these case programing will ignore this mes for pushing.*/
                        continue;
                    }
                    if(response == null){
                        /*Perhaps response is null for example: channel already
                        be released before client's quit command push.*/
                        continue;
                    }

                    //获取当前绑定的其余客户端.
                    List<Map.Entry<Object, SocketProxy>> otherBindClients =
                            this.getProxies().entrySet().stream().filter(kv -> {
                                return !kv.getKey().equals(mesItem.getRegisterInfo().getIdentity()) && kv.getValue().getState() == SocketState.BIND;
                            }).collect(Collectors.toList());
                    if(otherBindClients.size() == 0){
                        /*Current only exist one client, or all client has quit in a moment, ignore this mes.*/
                        continue;
                    }

                    //构建协议响应内容
                    CarFunProtocol protocol = null;
                    try {
                        protocol = Cast.cast($protocolBuilder.invoke(mesItem.getCommand(),response.getCode()),CarFunProtocol.class);
                    }catch (Throwable e){
                        /*反射异常，抛服务异常，这里只需要在保证编译阶段无异常基本就不会出现(除了内存堆栈的异常等.)*/
                        continue;
                    }
                    protocol.charSet(UTF_8);
                    if(response.getData() != null){
                        protocol.data(JSON.toJSONString(response.getData()).getBytes(UTF_8));
                        if(response.getData() instanceof List){
                            protocol.isJSONAry(true);
                        }
                    }
                    byte[] responseProtocol = protocol.build().toString().getBytes(UTF_8);

                    //通知到其余客户端
                    for (Map.Entry<Object, SocketProxy> otherBindClient : otherBindClients) {
                        try {
                            otherBindClient.getValue().writeToClient(responseProtocol);
                        }catch (IOException e){
                            /*Write mesInfo to clients which binding on current channel failed.
                            * perhaps socket client already been released, so server will destroy this client
                            * and release socket thread. */
                            otherBindClient.getValue().destroy();
                        }
                    }
                }else {
                    /*是否当一段时间内没有任何Mes时使线程sleep，sleep策略是怎么样的？*/
                    //Thread.sleep(1000);
                }
            }catch (Throwable e){
                /*出现了处理过程中的运行时异常，可能由于服务端线程/内存资源耗尽
                * 或者出现了Command断言的异常, 或者是连接channel的客户端均解绑
                * channel资源被释放等，输出异常信息，该message忽略.*/
                e.printStackTrace();
                LogUtil.error("Channel :: Throwable :: Developer check if anywhere" +
                        "throw RuntimeException and has none dealing!!!");
                continue;
            }
        }
    }



    /**
     * 客户端加入
     * @param creatorProxy
     * @param creatorRegisterInfo
     */
    public synchronized void clientJoin(@NotNull SocketProxy creatorProxy,
                                        @NotNull RegisterInfo creatorRegisterInfo){
        registerInfos.put(creatorRegisterInfo.getIdentity(),creatorRegisterInfo);
        proxies.put(creatorRegisterInfo.getIdentity(),creatorProxy);
        //添加客户端加入的消息
        push(creatorRegisterInfo,creatorProxy, EventCommand.EventChannel.EVENT_CHANNEL_JOIN,null);
    }

    /**
     * 客户端断链,取消在Channel中的注册信息.
     * @param socketProxy
     */
    public synchronized void clear(@NotNull SocketProxy socketProxy){
        //NotNull
        Object clearSocketProxy = this.proxies.remove(socketProxy.getRegisterInfo().getIdentity());
        //NotNull
        RegisterInfo remove = this.registerInfos.remove(socketProxy.getRegisterInfo().getIdentity());
        /*
            当channel线程中没有任何socketProxy连接时，立即销毁channel，释放channel线程,
            此时原本连接的socketProxy均处于free态，等待资源回收，如果此时某一断联的客户端在channel销毁后再进行加入，
            则加入失败, 组队已关闭.
         */
        if(this.proxies.size() == 0){
            this.state = ChannelState.DESTROY;
            this.mesPool.clear();
            MindManger.channels.remove(this.channelId);
        }else {
            //添加客户端退出的消息
            push(remove,socketProxy,EventCommand.EventChannel.EVENT_CHANNEL_QUIT,null);
        }
    }

    /**
     * Add change metadata to mesPool, which given from socket clients.
     * @param registerInfo
     * @param eventCommand
     * @param data
     */
    private synchronized void push(@NotNull RegisterInfo registerInfo,
                                        @NotNull SocketProxy proxy,
                                        @NotNull EventCommand eventCommand,
                                        @Nullable JSON data){
        PoolMesItem<JSON> poolMesItem = new PoolMesItem<>(registerInfo,proxy,eventCommand, data);
        this.mesPool.add(poolMesItem);
    }

    /**
     * Push by check proxy's state.
     * @param proxy
     * @param eventCommand
     * @param data
     * @throws CommandExecException
     */
    public void push(@NotNull SocketProxy proxy,
                     @NotNull EventCommand eventCommand,
                     @Nullable JSON data) throws CommandExecException{
        if(proxy == null || proxy.getState() != SocketState.BIND){
            throw new CommandExecException("Client has bot BIND anyone channel currently!");
        }
        push(proxy.getRegisterInfo(),proxy,eventCommand,data);
    }

    @Override
    public String toString() {
        return "Channel{" +
                "channelId=" + channelId +
                ", state=" + state +
                ", proxiesSize=" + String.valueOf(proxies != null ? proxies.size() : 0) +
                ", registerInfosSize=" + String.valueOf(registerInfos != null ? registerInfos.size() : 0) +
                ", mesPoolSize=" + mesPool.size() +
                '}';
    }

    public Long getChannelId() {
        return channelId;
    }

    public ChannelState getState() {
        return state;
    }

    public void setState(ChannelState state) {
        this.state = state;
    }

    public Map<Object, SocketProxy> getProxies() {
        return proxies;
    }

    public void setProxies(Map<Object, SocketProxy> proxies) {
        this.proxies = proxies;
    }

    public Map<Object, RegisterInfo> getRegisterInfos() {
        return registerInfos;
    }

    public void setRegisterInfos(Map<Object, RegisterInfo> registerInfos) {
        this.registerInfos = registerInfos;
    }

    public BlockingQueue<PoolMesItem<JSON>> getMesPool() {
        return mesPool;
    }

    public void setMesPool(BlockingQueue<PoolMesItem<JSON>> mesPool) {
        this.mesPool = mesPool;
    }

    public static MethodHandle get$protocolBuilder() {
        return $protocolBuilder;
    }

    public static Charset getUtf8() {
        return UTF_8;
    }
}
